package dse.messages.multiversion;

import java.util.LinkedList;
import java.util.Map;
import java.util.TreeMap;
import java.util.TreeSet;

import dse.DataObject;
import dse.Experiment;
import dse.ExperimentRecorder;
import dse.Node;
import dse.Version;
import dse.messages.Message;
import dse.messages.MessagePurpose;
import dse.messages.MessageUtil;
import dse.messages.Transaction;
import dse.messages.singleversion.ShareDownloader;

public class GetGlobalDhtCVSTransaction extends Transaction implements
		ShareDownloader, VersionReader {
	private Node requester;
	private DataObject desiredObject;
	private LinkedList<Node> residentServers;
	private LinkedList<Node> nodesToBeRepaired;
	private boolean failed;
	private TreeMap<Version, LinkedList<Node>> sharesDownloaded;
	private Version version;
	private Node versionServer;
	private int requestShareCount;
	private int validShareCount;
	private int sharesToBeUpdate;
	private int maxFailure;
	private int successCount;
	private int failureCount;
	
	public GetGlobalDhtCVSTransaction (
			Experiment experiment,
			int id,
			Node requester,
			DataObject desiredObject,
			double time
			)
	{
		super(id, requester, experiment, time);
				
		this.requester = requester;
		this.desiredObject = desiredObject;
		this.failed = false;
		this.validShareCount = 0;
		this.sharesToBeUpdate = desiredObject.shareCount();
		if(isUpdate && isPartialUpdate)
		{
			this.sharesToBeUpdate = (int)(desiredObject.shareCount()*Math.random());
			if(this.sharesToBeUpdate == 0)
				this.sharesToBeUpdate = 1;
		}
		this.maxFailure = experiment.modules().nodeEventScheduler().getEventsPerPeriod();
		this.successCount = 0;
		this.failureCount = 0;
		this.nodesToBeRepaired = null;
		this.sharesDownloaded = null;
	}
	
	@Override
	public void start(ExperimentRecorder recorder) {
		// TODO Auto-generated method stub
		residentServers = MessageUtil.getNodesByDht(experiment.network(), requester, desiredObject, experiment.modules().assigner());		
		versionServer = experiment.modules().assigner().assign(desiredObject, 1, experiment.network());
		
		if(isUpdate)
		{
			recorder.dataUpdateInitiated(super.id, requester, desiredObject, startTime);
			recorder.debug(this, "start", startTime);
		
			validShareCount = desiredObject.shareCount()-this.maxFailure;
			version = new Version(requester.id(), requester.sequence());
			sendVersionRequest(recorder, versionServer, startTime);
		}
		else
		{
			recorder.dataRequestInitiated(super.id, requester, desiredObject, startTime);
			recorder.debug(this, "start", startTime);
			
			this.requestShareCount = desiredObject.requiredShareCount()+this.maxFailure;
			this.validShareCount = desiredObject.requiredShareCount();
			sendVersionRequest(recorder, versionServer, startTime);
		}
	}
	
	private void sendVersionRequest(ExperimentRecorder recorder, Node node, Double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestVersionTLOMessage(
				requester, node, desiredObject, isUpdate?MessagePurpose.DataObjectUpdate:MessagePurpose.DataObjectRequest,
				isUpdate, residentServers, this);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
		
	private void sendUpdateShareRequest(ExperimentRecorder recorder, Node node, Version version, boolean trx, boolean repair, Double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createShareUpdateCVSMessage(requester, node, desiredObject,
				version, MessagePurpose.DataObjectUpdate, this);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}

	@Override
	public boolean isDoneDownloadingShares() {
		// TODO Auto-generated method stub
		return failed || finished;
	}

	@Override
	public boolean isDoneUpdatingShares() {
		// TODO Auto-generated method stub
		return failed || finished;
	}

	@Override
	public void shareDownloadFailed(ExperimentRecorder recorder,
			DataObject dobj, int shareIndex, double time) {
		// TODO Auto-generated method stub

	}

	@Override
	public void shareDownloaded(ExperimentRecorder recorder, Node from,
			DataObject dobj, int shareIndex, double time) {
		// TODO Auto-generated method stub
	}

	@Override
	public void shareUpdateFailed(ExperimentRecorder recorder, DataObject dobj,
			int shareIndex, double time) {
		// TODO Auto-generated method stub
		if(isUpdate) {
			this.failureCount++;
			if((successCount + failureCount) >= this.sharesToBeUpdate)
			{
				if(this.successCount >= validShareCount)
					transactionSucceed(recorder, time);
				else
					transactionFailed(recorder, time);
			}
		}
	}

	@Override
	public void shareUpdated(ExperimentRecorder recorder, Node to,
			DataObject dobj, int shareIndex, double time) {
		// TODO Auto-generated method stub
		if(isUpdate) {		
			//this.processedUpdateMessage++;
			//this.processedSuccessUpdateMessage++;
			this.successCount++;
			if((successCount+failureCount) >= this.sharesToBeUpdate)
			{
				if(this.successCount >= validShareCount)
					transactionSucceed(recorder, time);
				else
					transactionFailed(recorder, time);
			}
		}
	}

	@Override
	public void readVersion(ExperimentRecorder recorder,
			TreeSet<Version> versions, double time, Node from) {}
	
	@Override
	public void readVersion(ExperimentRecorder recorder, int seq, int index, double time){}


	@Override
	public void readVersion(ExperimentRecorder recorder, Node from,
			Version version, double time) {
		// TODO Auto-generated method stub
		if(finished)
			return;
		if(version == null)
		{
			failureCount++;
			if(this.nodesToBeRepaired == null){
				this.nodesToBeRepaired = new LinkedList<Node>();
			}
			this.nodesToBeRepaired.add(from);
			
			if(successCount+failureCount>=this.requestShareCount)
				processVersion(recorder, time);
			return;
		}
		
		successCount++;
		if(sharesDownloaded == null)
			sharesDownloaded = new TreeMap<Version, LinkedList<Node>>();
		if(this.sharesDownloaded.get(version) == null)
		{
			this.sharesDownloaded.put(version, new LinkedList<Node>());
		}
		this.sharesDownloaded.get(version).add(from);
		
		if(successCount+failureCount>=this.requestShareCount)
			processVersion(recorder, time);
	}

	@Override
	public void readVersion(ExperimentRecorder recorder, int seq,
			double time) {
		// TODO Auto-generated method stub
		if(seq == -1)
			return;
		version = new Version(seq+1, requester.id());
		int count = 0;
		for(Node n : this.residentServers){
            this.sendUpdateShareRequest(recorder, n, version, true, false, time);
            count++;
            if(count == this.sharesToBeUpdate)
            	break;
        }
	}
	
	@Override
	public void readVersion(ExperimentRecorder recorder, double time, LinkedList<Version> versions, Version version, Node from){}
	
	private void processVersion(ExperimentRecorder recorder, double time)
	{
		if((sharesDownloaded==null) || (sharesDownloaded.size()==0)){
			transactionFailed(recorder, time);
			return;
		}
		
		Map.Entry<Version, LinkedList<Node>> vn = sharesDownloaded.pollLastEntry();
		if(vn.getValue().size() == requestShareCount)
		{
			transactionSucceed(recorder, time);
		}
		else if(vn.getValue().size() < requestShareCount && vn.getValue().size()>=validShareCount)
		{
			version = vn.getKey();
			for(Node node : this.nodesToBeRepaired)
			{
				sendUpdateShareRequest(recorder, node, version, false, true, time);
			}
			transactionSucceed(recorder, time);	
		}
		else
			transactionFailed(recorder, time);		
	}

	private void transactionSucceed(ExperimentRecorder recorder, double time)
	{
		finished = true;
		failed  = false;
		//System.out.println("Data ID : " + desiredObject.id() + "Version : " + maxVersion.getSequence() + " , " + maxVersion.getClientID());
		if(isUpdate)
		{
			recorder.debug(
					"GetDhtDataSharesWithVersionTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataUpdateFinished(super.id, requester, desiredObject, true, time);
		}
		else
		{
			recorder.debug(
					"GetDhtDataSharesWithVersionTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			/*for(Node node : this.residentServers.keySet())
			{
				Message m = super.createReadCompleteMessage(requester, node, desiredObject, MessagePurpose.DataObjectRequest, this);
				recorder.debug(this, "new msg: " + m, time);
				experiment.messageQueue().enqueueMessage(m, time);
			}*/
			Message m = super.createReadCompleteTLOMessage(requester, versionServer, desiredObject, true, MessagePurpose.DataObjectRequest, this);
			recorder.debug(this, "new msg: " + m, time);
			experiment.messageQueue().enqueueMessage(m, time);
			recorder.dataRequestFinished(super.id, requester, desiredObject, true, this.residentServers.get(this.validShareCount-1),
					false, time);
		}
	}
	
	private void transactionFailed (ExperimentRecorder recorder, double time)
	{
		recorder.debug(this, "transFailed", time);
		
		finished = true;
		failed = true;
		if(isUpdate)
			recorder.dataUpdateFinished(super.id, requester, desiredObject, false, time);
		else{
			/*for(Node node : this.residentServers.keySet())
			{
				Message m = super.createReadCompleteMessage(requester, node, desiredObject, MessagePurpose.DataObjectRequest, this);
				recorder.debug(this, "new msg: " + m, time);
				experiment.messageQueue().enqueueMessage(m, time);
			}*/
			Message m = super.createReadCompleteTLOMessage(requester, versionServer, desiredObject, true, MessagePurpose.DataObjectRequest, this);
			recorder.debug(this, "new msg: " + m, time);
			experiment.messageQueue().enqueueMessage(m, time);
			recorder.dataRequestFinished(super.id, requester, desiredObject, false, null,
				false, time);
		}
	}
	public Transaction getTransaction(){
		return this;
	}
}
